package io.github.bigbird0101.datatransfer.exchanger;

import integration.data.task.api.transformation.impl.TransformerExecution;
import integration.modle.exception.IntegrationValidException;
import integration.utils.trace.LogUtils;
import io.github.bigbird0101.datatransfer.collector.Communication;
import io.github.bigbird0101.datatransfer.collector.CommunicationTool;
import io.github.bigbird0101.datatransfer.collector.TaskPluginCollector;
import io.github.bigbird0101.datatransfer.record.Record;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

@Slf4j
@Setter
@Getter
public abstract class TransformerExchanger {

    protected String taskId;
    protected Communication currentCommunication;
    protected TaskPluginCollector pluginCollector;

    private List<TransformerExecution> transformerExecs;


    private long totalExaustedTime = 0;
    private long totalFilterRecords = 0;
    private long totalSuccessRecords = 0;
    private long totalFailedRecords = 0;

    public TransformerExchanger() {
    }

    public TransformerExchanger(String taskId, Communication communication,
                                List<TransformerExecution> transformerExecs,
                                final TaskPluginCollector pluginCollector) {
        this.transformerExecs = transformerExecs;
        this.pluginCollector = pluginCollector;
        this.taskId = taskId;
        this.currentCommunication = communication;
    }

    public Record doTransformer(Record record) {
        if (transformerExecs == null || transformerExecs.size() == 0) {
            return record;
        }

        Record result = record;

        long diffExaustedTime = 0;
        String errorMsg = null;
        boolean failed = false;
        for (TransformerExecution transformerInfoExec : transformerExecs) {
            long startTs = System.nanoTime();
            /**
             * 延迟检查transformer参数的有效性，直接抛出异常，不作为脏数据
             * 不需要在插件中检查参数的有效性。但参数的个数等和插件相关的参数，在插件内部检查
             */
            if (!transformerInfoExec.isChecked()) {

                if (transformerInfoExec.getColumnIndex() != null && transformerInfoExec.getColumnIndex() >= record.getColumnNumber()) {
                    throw new IntegrationValidException(String.format("转换函数参数异常：columnIndex[%s] out of bound[%s]. name=%s",
                            transformerInfoExec.getColumnIndex(), record.getColumnNumber(),
                            transformerInfoExec.getTransformerName()));
                }
                transformerInfoExec.setIsChecked(true);
            }

            try {
                // 执行转换函数处理
                result = transformerInfoExec.getTransformer().evaluate(result, transformerInfoExec.getParas());
            } catch (Exception e) {
                errorMsg = String.format("transformer(%s) has Exception(%s)", transformerInfoExec.getTransformerName(),
                        e.getMessage());
                failed = true;
                LogUtils.error(log, errorMsg, e);
                //脏数据不再进行后续transformer处理，按脏数据处理，并过滤该record。
                break;
            }

            if (result == null) {
                /**
                 * 这个null不能传到writer，必须消化掉
                 */
                totalFilterRecords++;
                log.debug("被过滤的数据：{}", record);
                break;
            }

            long diff = System.nanoTime() - startTs;
            diffExaustedTime += diff;
        }

        totalExaustedTime += diffExaustedTime;
        if (failed) {
            totalFailedRecords++;
            // 脏数据收集
            this.pluginCollector.collectDirtyRecord(record, errorMsg);
            return null;
        } else {
            totalSuccessRecords++;
            return result;
        }
    }

    public void doStat() {

        if (currentCommunication != null) {
            currentCommunication.setLongCounter(CommunicationTool.TRANSFORMER_SUCCEED_RECORDS, totalSuccessRecords);
            currentCommunication.setLongCounter(CommunicationTool.TRANSFORMER_FAILED_RECORDS, totalFailedRecords);
            currentCommunication.setLongCounter(CommunicationTool.TRANSFORMER_FILTER_RECORDS, totalFilterRecords);
            currentCommunication.setLongCounter(CommunicationTool.TRANSFORMER_USED_TIME, totalExaustedTime);
        }
    }

}
